package cn.callmee.springboot.pulsar.starter.client.config;

import cn.callmee.springboot.pulsar.starter.client.annotations.PulsarConsumer;
import cn.callmee.springboot.pulsar.starter.client.annotations.PulsarProducer;
import cn.callmee.springboot.pulsar.starter.client.domain.PulsarUrlGenerator;
import cn.callmee.springboot.pulsar.starter.client.holder.Holder;
import cn.callmee.springboot.pulsar.starter.client.message.FailedMessage;
import cn.callmee.springboot.pulsar.starter.client.message.PulsarMessage;
import cn.callmee.springboot.pulsar.starter.client.properties.PulsarProperties;
import cn.callmee.springboot.pulsar.starter.client.utils.SchemaUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.ImmutableTriple;
import org.apache.pulsar.client.api.*;
import org.slf4j.helpers.MessageFormatter;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.EmbeddedValueResolverAware;
import org.springframework.context.annotation.DependsOn;
import org.springframework.context.annotation.Import;
import org.springframework.context.event.EventListener;
import org.springframework.util.StringValueResolver;
import reactor.core.publisher.Sinks;
import reactor.util.concurrent.Queues;

import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Import(PulsarUrlGenerator.class)
public abstract class PulsarClientInitial implements BeanPostProcessor, EmbeddedValueResolverAware {

    protected final Sinks.Many<FailedMessage> SINK = Sinks.many().multicast().onBackpressureBuffer(Queues.SMALL_BUFFER_SIZE, false);
    protected static final Map<String, ImmutableTriple<String, PulsarProducer, Producer>> PRODUCERS = new ConcurrentHashMap<>();
    protected static final Map<String, ImmutableTriple<String, PulsarConsumer, Holder.ConsumerHolder>> CONSUMERS = new ConcurrentHashMap<>();

    protected static LinkedList<String> REG_PRODUCER_LOG_ARRAY = new LinkedList<>();
    protected static LinkedList<String> REG_CONSUMER_LOG_ARRAY = new LinkedList<>();

    public static StringValueResolver STRING_VALUE_RESOLVER;
    protected static List<Consumer> COLLECT_CONSUMERS = new ArrayList<>();
    @Autowired
    protected PulsarClient pulsarClient;
    @Autowired
    protected PulsarUrlGenerator pulsarUrlGenerator;
    @Resource
    protected PulsarProperties pulsarProperties;
    // protected ProducerInterceptor producerInterceptor;
    // protected ConsumerInterceptor consumerInterceptor;

    @EventListener(ApplicationReadyEvent.class)
    @DependsOn({"pulsarOauth2Properties", "pulsarTlsProperties", "pulsarConsumerProperties"})
    public void init() {
        log.info("\n[CONFIG]注册消息生产者: \n{}", String.join("\n", PulsarClientInitial.REG_PRODUCER_LOG_ARRAY));
        PulsarClientInitial.REG_PRODUCER_LOG_ARRAY = null;
        log.info("\n[CONFIG]注册消息消费者: \n{}", String.join("\n", PulsarClientInitial.REG_CONSUMER_LOG_ARRAY));
        PulsarClientInitial.REG_CONSUMER_LOG_ARRAY = null;
        // log.info("ProducerInterceptor:{}", producerInterceptor);
        // log.info("ConsumerInterceptor:{}", consumerInterceptor);
    }

    protected void addInLog(LinkedList<String> logArray, String pattern, Object[] objects) {
        logArray.add(MessageFormatter.arrayFormat(pattern, objects).getMessage());
    }

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        Class<?> beanClass = bean.getClass();
        Arrays.stream(beanClass.getDeclaredMethods())
                .forEach($ -> {
                    if ($.isAnnotationPresent(PulsarProducer.class)) {
                        postInitializationProducer($, beanClass);
                    }
                    if ($.isAnnotationPresent(PulsarConsumer.class)) {
                        postInitializationConsumer($, bean, beanClass);
                    }
                });
        return bean;
    }

    protected abstract void postInitializationProducer(Method $, Class<?> beanClass);

    protected abstract void postInitializationConsumer(Method $, Object bean, Class<?> beanClass);

    protected <T> Schema<?> getSchema(Holder.ProducerHolder holder) throws RuntimeException {
        return SchemaUtils.getSchema(holder.getSerialization(), holder.getClazz());
    }

    public ImmutableTriple<String, PulsarProducer, Producer> getProducer(String topic) {
        return PRODUCERS.get(STRING_VALUE_RESOLVER.resolveStringValue(topic));
    }

    @Override
    public void setEmbeddedValueResolver(StringValueResolver stringValueResolver) {
        STRING_VALUE_RESOLVER = stringValueResolver;
    }

    protected <T> PulsarMessage<T> wrapMessage(Message<T> message) {
        final PulsarMessage<T> pulsarMessage = new PulsarMessage<T>();

        pulsarMessage.setValue(message.getValue());
        pulsarMessage.setMessageId(message.getMessageId());
        pulsarMessage.setSequenceId(message.getSequenceId());
        pulsarMessage.setProperties(message.getProperties());
        pulsarMessage.setTopicName(message.getTopicName());
        pulsarMessage.setKey(message.getKey());
        pulsarMessage.setEventTime(message.getEventTime());
        pulsarMessage.setPublishTime(message.getPublishTime());
        pulsarMessage.setProducerName(message.getProducerName());

        return pulsarMessage;
    }
}

